package com.leigang.rocketmq.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * @author leig-a
 * @version 1.0
 * @desc
 * @date 2022/6/15 16:22
 */
@Slf4j
@Component
public class RocketConsumer implements CommandLineRunner {

    @Value("${apache.rocketmq.consumer.pushConsumer}")
    private String pushConsumer;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @Value("${message.first.topic}")
    private String topicFirst;

    @Value("${message.first.tag}")
    private String tagFirst;

    private static final String ROCKET_GROUP = "SpringBootRocketMqGroup";

    /**
     * 初始化RocketMq的监听信息，渠道信息
     */
    public void messageListener(String topicStr, String tagStr) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ROCKET_GROUP);
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            // 订阅PushTopic下Tag为push的消息,都订阅消息
            consumer.subscribe(topicStr, tagStr);
            // 程序第一次启动从消息队列头获取数据
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //可以修改每次消费消息的数量，默认设置是每次消费一条
            consumer.setConsumeMessageBatchMaxSize(1);
            //在此监听中消费信息，并返回消费的状态信息
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                // 会把不同的消息分别放置到不同的队列中
                for (Message msg : msgs) {
                    log.info("messageListener msg: {}", new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
        } catch (Exception e) {
            log.info("messageListener Exception : {}", e.getMessage());
            e.printStackTrace();
        }
    }

    @Override
    public void run(String... args) {
        this.messageListener(topicFirst, tagFirst);
    }
}
